/*
 * Copyright 2007 Stefan Arentz <http://stefan.arentz.nl>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.sateh.spaces.simple;

import com.sateh.spaces.simple.matcher.ObjectMatcher;
import com.sateh.spaces.simple.matcher.SimpleSpaceObjectClassMatcher;
import com.sateh.spaces.simple.matcher.SimpleSpaceObjectTemplateMatcher;
import com.sateh.spaces.simple.store.SimpleSpaceObjectStore;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;

/**
 * This class implements the space implementation. It is Stefan's  first naive try to implement his own flavour of
 * the JavaSpaces specification.
 *
 * @author Stefan Arentz <stefan@arentz.nl>
 */

public class SimpleSpace
{
    private final static Log log = LogFactory.getLog(SimpleSpace.class);

    /**
     * 
     */

    private UUID id = UUID.randomUUID();
    
    /**
     * 
     */

    public SimpleSpace()
    {
    }

    /**
     *
     */

    public UUID getId()
    {
        return id;
    }

    /**
     *
     */

    private final Set<SimpleSpaceObjectWaiter> objectWaiters = Collections.synchronizedSet(new HashSet<SimpleSpaceObjectWaiter>());
    
    /**
     *
     */
    
    private final SimpleSpaceObjectStore objectStore = new SimpleSpaceObjectStore();

    /**
     * According to The JavaSpaces Service Specification:
     *
     * An entry that is written is not visible outside its transaction until the transaction successfully commits. If
     * the entry is taken within the transaction, the entry will never be visible outside the transaction and will not
     * be added to the space when the transaction commits. Specifically, the entry will not generate notifications to
     * listeners that are not registered under the writing transaction. Entries written under a transaction that aborts
     * are discarded.
     */

    // TODO Add the transactionId to the ObjectWaiter?

    UUID writeSpaceObject(SimpleSpaceObject object, long duration, SimpleSpaceTransactionId transactionId)
    {
        boolean objectNeedsToGoInSpace = true;

        // First see if there was an object waiter waiting for this object. If so then the object
        // never really enters the space, we simply push it back to the one waiting for it.

        synchronized(objectWaiters)
        {
            for (SimpleSpaceObjectWaiter waiter : objectWaiters)
            {
                if (waiter.getMatcher().willMatch(object))
                {
                    waiter.put(object);
                    objectWaiters.remove(waiter);

                    // Don't put the object in the space if a 'taker' was waiting for it

                    if (waiter.getType() == SimpleSpaceObjectWaiter.TAKER) {
                        objectNeedsToGoInSpace = false;
                    }

                    break;
                }
            }
        }
        
        // If there was no waiter for this object then add it to the space.

        if (objectNeedsToGoInSpace) {
            objectStore.storeObject(object, transactionId);
        }

        // Return the object id no matter what. Even if the object is taken directly by a waiter, we
        // still return a Lease object containing this object id to the client. TODO Is that right?

        return object.getId();
    }

    /**
     *
     */

    public SimpleSpaceObject takeOrWaitForSpaceObject(String objectClass, long timeout, SimpleSpaceTransactionId transactionId)
    {
        // First we see if there is an object of the requested class available. This is not very
        // efficient but we can always refactor that later on.

        ObjectMatcher matcher = new SimpleSpaceObjectClassMatcher(objectClass, transactionId);

        SimpleSpaceObject object = objectStore.matchAndTakeObject(matcher);
        if (object != null) {
            return object;
        }

        // If not then we create a object waiter. The waiter is a combination of a matcher and a
        // blocking queue. The main object adder thread will push all objects through all the
        // registered matchers and in case of a succesful willMatch unlock the wait on the queue so
        // that the object will be retrieved by the waiter.

        SimpleSpaceObjectWaiter waiter = new SimpleSpaceObjectWaiter(matcher, SimpleSpaceObjectWaiter.TAKER);
        objectWaiters.add(waiter);

        // Wait for an object to become available.
        // TODO Can this be refactored with goodness from java.util.concurrent?

        return waiter.poll(timeout);
    }

    /**
     *
     */

    public SimpleSpaceObject takeSpaceObject(String objectClass, SimpleSpaceTransactionId transactionId)
    {
        ObjectMatcher matcher = new SimpleSpaceObjectClassMatcher(objectClass, transactionId);
        return objectStore.matchAndTakeObject(matcher);
    }

    /**
     *
     */

    public SimpleSpaceObject takeSpaceObject(String objectClass, SimpleSpaceTemplate template, SimpleSpaceTransactionId transactionId)
    {
        ObjectMatcher matcher = new SimpleSpaceObjectTemplateMatcher(template, transactionId);
        return objectStore.matchAndTakeObject(matcher);
    }

    /**
     *
     */

    public SimpleSpaceObject takeOrWaitForSpaceObject(String objectClass, SimpleSpaceTemplate templateProperties, long timeout, SimpleSpaceTransactionId transactionId)
    {
        // First we see if there is an object of the requested class available. This is not very
        // efficient but we can always refactor that later on.

        ObjectMatcher matcher = new SimpleSpaceObjectTemplateMatcher(templateProperties, transactionId);

        SimpleSpaceObject object = objectStore.matchAndTakeObject(matcher);
        if (object != null) {
            return object;
        }

        // If not then we create a object waiter. The waiter is a combination of a matcher and a
        // blocking queue. The main object adder thread will push all objects through all the
        // registered matchers and in case of a succesful willMatch unlock the wait on the queue so
        // that the object will be retrieved by the waiter.

        SimpleSpaceObjectWaiter waiter = new SimpleSpaceObjectWaiter(matcher, SimpleSpaceObjectWaiter.TAKER);
        objectWaiters.add(waiter);

        // Wait for an object to become available.
        // TODO Can this be refactored with goodness from java.util.concurrent?

        return waiter.poll(timeout);
    }

    //
    // A read may match any entry written under that transaction or in the entire space. A JavaSpaces service is not
    // required to prefer matching entries written inside the transaction to those in the entire space. When read, an
    // entry is added to the set of entries read by the provided transaction. Such an entry may be read in any other
    // transaction by which the entry is visible, but cannot be taken in another transaction.
    //

    /**
     * 
     */

    public SimpleSpaceObject readOrWaitForSpaceObject(String className, long timeout, SimpleSpaceTransactionId transactionId)
    {
        // First we check if the object is readily available in the object store.

        ObjectMatcher matcher = new SimpleSpaceObjectClassMatcher(className, transactionId);

        SimpleSpaceObject object = objectStore.matchAndReadObject(matcher);
        if (object != null) {
            return object;
        }

        // If not then we create a object waiter. The waiter is a combination of a matcher and a
        // blocking queue. The main object adder thread will push all objects through the matcher
        // and in case of a willMatch unlock the wait on the queue.

        SimpleSpaceObjectWaiter waiter = new SimpleSpaceObjectWaiter(matcher, SimpleSpaceObjectWaiter.READER);
        objectWaiters.add(waiter);

        // Wait for an object to become available.

        return waiter.poll(timeout);
    }
    
    /**
     *
     */

    public SimpleSpaceObject readSpaceObject(String objectClass, SimpleSpaceTransactionId transactionId)
    {
        ObjectMatcher matcher = new SimpleSpaceObjectClassMatcher(objectClass, transactionId);
        return objectStore.matchAndReadObject(matcher);
    }

    /**
     *
     */

    public SimpleSpaceObject readSpaceObject(String objectClass, SimpleSpaceTemplate template, SimpleSpaceTransactionId transactionId)
    {
        ObjectMatcher matcher = new SimpleSpaceObjectTemplateMatcher(template, transactionId);
        return objectStore.matchAndReadObject(matcher);
    }

    /**
     *
     */

    public SimpleSpaceObject readOrWaitForSpaceObject(SimpleSpaceTemplate template, long timeout, SimpleSpaceTransactionId transactionId)
    {
        // First we see if there is an object of the requested class available. This is not very
        // efficient but we can always refactor that later on.

        ObjectMatcher matcher = new SimpleSpaceObjectTemplateMatcher(template, transactionId);

        SimpleSpaceObject object = objectStore.matchAndReadObject(matcher);
        if (object != null) {
            return object;
        }

        // If not then we create a object waiter. The waiter is a combination of a matcher and a
        // blocking queue. The main object adder thread will push all objects through the matcher
        // and in case of a willMatch unlock the wait on the queue.

        SimpleSpaceObjectWaiter waiter = new SimpleSpaceObjectWaiter(matcher, SimpleSpaceObjectWaiter.READER);
        objectWaiters.add(waiter);

        // Wait for an object to become available.

        return waiter.poll(timeout);
    }
    
    /**
     *
     */
    
    public boolean cancelLeaseForObjectWithId(UUID objectId)
    {
        return objectStore.cancelObjectLease(objectId);
    }

    /**
     *
     */
    
    public boolean renewLeaseForObjectWithId(UUID objectId, long timeToLive)
    {
        return objectStore.renewObjectLease(objectId, timeToLive);
    }

    //

    /**
     *
     */

    public long count(String className, SimpleSpaceTransactionId transactionId)
    {
        ObjectMatcher matcher = new SimpleSpaceObjectClassMatcher(className, transactionId);
        return objectStore.matchAndCount(matcher, transactionId);
    }

    /**
     *
     */
    
    public long count(SimpleSpaceTemplate template, SimpleSpaceTransactionId transactionId)
    {
        ObjectMatcher matcher = new SimpleSpaceObjectTemplateMatcher(template, transactionId);
        return objectStore.matchAndCount(matcher, transactionId);
    }

    //

    /**
     *
     */

    public void commitTransaction(SimpleSpaceTransactionId transactionId)
    {
        objectStore.mergeObjectsInTransaction(transactionId);
    }

    /**
     *
     */

    public void rollbackTransaction(SimpleSpaceTransactionId transactionId)
    {
        objectStore.releaseObjectsInTransaction(transactionId);
    }
}
